package com.intct.ads;

import com.intct.ods.OdsManApp;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema;
import org.apache.doris.flink.source.DorisSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

/**
 * @author gufg
 * @since 2025-08-21 09:59
 *
 * 1、原始数据：
 * {
 *   "driver_name": "金d488743e691a15e9c8f5c6e59b2cfb9b",
 *   "cancel_count": "0",
 *   "driver_id": "9",
 *   "lng": "104.10792",
 *   "create_time": "2019-04-18 06:47:14",
 *   "driver_type": "2",
 *   "mobile": "1772***3863",
 *   "register_city": "420300",
 *   "time": "1475947101",
 *   "order_id": "17592719330238",
 *   "lat": "30.67518",
 *   "driver_management_id": "0"
 * }
 *
 *
 * {
 *   "driver_id": "end",
 *   "order_id": "17592719302995"
 * }
 *
 * 2、目标：保存HBase(保存历史)、Redis(前端展示)

 * 3、存在需要解决问题
 *      3.1 数据顺序：
 *          3.1.1 正序  -- 没问题
 *          3.1.2 乱序  -- 有问题
 *      3.2 订单结束状态
 *
 * 4、解决方案
 *      4.1 数据顺序(乱序)：
 *          Flink : 水位线+时间+窗口
 *
 *          Redis : 数据存储到有顺序集合类型：键(订单) 分数(时间戳) 值(经度+":"+纬度)
 *              司机：李四  订单：21123123  时间戳：1717171717 经度：123 纬度：20   --> 起点
 *              司机：李四  订单：21123123  时间戳：1717171718 经度：124 纬度：20   --> 1
 *              司机：李四  订单：21123123  时间戳：1717171719 经度：124 纬度：20   --> 2
 *              司机：李四  订单：21123123  时间戳：1717171720 经度：124 纬度：20   --> 3
 *              司机：李四  订单：21123123  时间戳：1717171721 经度：124 纬度：20   --> 4
 *              司机：李四  订单：21123123  时间戳：1717171722 经度：124 纬度：20   --> 5

 *              司机：李四  订单：21123123  时间戳：1717171717 经度：123 纬度：20   --> 起点
 *              司机：李四  订单：21123123  时间戳：1717171720 经度：124 纬度：20   --> 3
 *              司机：李四  订单：21123123  时间戳：1717171721 经度：124 纬度：20   --> 4
 *              司机：李四  订单：21123123  时间戳：1717171722 经度：124 纬度：20   --> 5
 *              司机：李四  订单：21123123  时间戳：1717171718 经度：124 纬度：20   --> 1
 *              司机：李四  订单：21123123  时间戳：1717171719 经度：124 纬度：20   --> 2

 *      4.2 订单结束状态
 * {
 *   "driver_id": "end",
 *   "order_id": "17592719302995"
 * }
 *          Redis : 数据存储到有顺序集合类型：键(订单) 分数(时间戳--9999999999) 值(经度+":"+纬度)
 */
public class AdsLocusApp {
    public static void main(String[] args) throws Exception {
        // 从命令中获取参数
        ParameterTool paraTools = ParameterTool.fromArgs(args);
        String mvnProjectPath = OdsManApp.class.getResource("/test.properties").getFile();

        String filePath = paraTools.get("filePath", mvnProjectPath);
        String topic = paraTools.get("kafak-topic-dwd-locus", "dwd_locus");

        // 从配置文件中获取配置参数
        ParameterTool propertiesFile = ParameterTool.fromPropertiesFile(filePath);

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置检查
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        DorisOptions option =
                DorisOptions.builder()
                        .setFenodes("180.76.99.167:8030")
                        .setTableIdentifier("intct.example_log")
                        .setUsername("root")
                        .setPassword("")
                        .build();

        DorisReadOptions readOptions =
                DorisReadOptions.builder()
                        .build();

        DorisSource.<List<?>>builder()
                .setDeserializer(new SimpleListDeserializationSchema())
                .build();

        DorisSource<List<?>> dorisSource =
                DorisSource.<List<?>>builder()
                        .setDorisOptions(option)
                        .setDorisReadOptions(readOptions)
                        .setDeserializer(new SimpleListDeserializationSchema())
                        .build();

        env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "doris source").print();

//        // 配置数据源
//        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(propertiesFile, topic, "ads_locus_group_id");
//        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "ads_locus_group_id");
//
//        // 数据处理
//        SingleOutputStreamOperator<LoucsBean> mapDS = sourceDS.map(new MapFunction<String, LoucsBean>() {
//            @Override
//            public LoucsBean map(String jsonStr) throws Exception {
//
//                // 从json string -> java bean
//                LoucsBean loucsBean = JSON.parseObject(jsonStr, LoucsBean.class);
//
//                String driverId = loucsBean.getDriverId();
//
//                // 订单结束
//                if ("end".equals(driverId)) {
//                    loucsBean.setTime(9999999999L);
//                    loucsBean.setLat("end");
//                    loucsBean.setLng("end");
//                }
//
//                return loucsBean;
//            }
//        });
//
//        // 输出数据--Redis
//        // zadd key_name 分数 值
//        mapDS.addSink(new AdsLocusRedisSink());
//
//        // 输出数据--HBase
//        mapDS.map(new AdsLocusFunc()).addSink(new HBaseSink("intct83", 6));

        // 启动作业
        env.execute();
    }
}
